package com.hyzs.spark.sql

/**
  * Created by Administrator on 2018/2/27.
  */
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.apache.spark.ml.feature.{MinMaxScaler, VectorAssembler}
import org.apache.spark.mllib.linalg.{Matrices, Vector}
import com.hyzs.spark.utils.SparkUtils._

object LabelDataProcess {
  val originalKey = "user_id"
  val key = "user_id_md5"
  import sqlContext.implicits._

  def importLabelTable(filePath:String):DataFrame = {
    val header = Array("phone", key, "label")
    val labelAll = createDFfromRawCsv(header, filePath, "\\t")
    saveTable(labelAll, "label_all")
    labelAll
  }

  // generate multiple label table
  def multiLabelProcess(labelRange:Int, labelTable:DataFrame): Unit = {
    for(index <- 1 to labelRange){
      val label_i = labelTable.select(
        col(key),
        when($"label" === s"$index","1").otherwise("0").as("label"))
      saveTable(label_i, s"label_$index")
    }
  }

  def main(args: Array[String]): Unit = {
    val resTable = Option(sc.getConf.get("spark.processJob.resultTable")).getOrElse("all_data")
    if(args.length>0 && args(0)=="import_label"){
      val all_label = importLabelTable("/hyzs/files/pin7labels.txt")
      multiLabelProcess(7, all_label)
    }



    if(args.length>1 && args(1)=="train"){

    }


  }

}
